#!/usr/bin/env io

/*
    Antipaxi - A possible implementation of the Paxos algorithm in Io.    
    Copyright (C) 2007 Darrin Eden

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License along
    with this program; if not, write to the Free Software Foundation, Inc.,
    51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/

hosts := list("site1", "site2", "site3")
domain := "example.com"
port := 4200
debug := false
hosts mapInPlace(host, "#{host}.#{domain}" interpolate)
hosts mapInPlace(host, 
	"#{DNSResolver ipForHostName(host)}:#{port}" interpolate)
defaultValue := SystemCall clone setCommand("hostname") run stdout readLine
proposals := List clone
hosts size repeat(proposals append(Object clone do(
	proposalNumber := 0;
	value := defaultValue; 
	highAccepted := 0;
	live := false)))
peers := Map clone addKeysAndValues(hosts, proposals)
majority := ((peers size/2) + 0.5) ceil
out := method(s, debug ifTrue("> #{s}" interpolate println); 
	coroDo(try(URL with(s) fetch)))

Proposer := Object clone do(
	prepare := method(peers foreach(peer, p, 
		s := "http://#{peer}/prepare/#{p proposalNumber}" interpolate;
		out(s)))
	ack := method(responded, highAccepted, highValue, peer, 
		p := peers at("#{peer}:#{port}" interpolate);
		p proposalNumber = responded; p live = true; 
		p highAccepted = highAccepted; 
		if(highValue == "none", p value = defaultValue, p value = highValue);
		a := 0; peers values foreach(q, 
			(q proposalNumber == responded and q live) ifTrue(a = a + 1); 
			(a >= majority) ifTrue(accept(responded))))
	accept := method(responded, 
		max := peers values foreach(x, List clone append(x highAccepted)) max;
		peers values foreach(x, 
			(x highAccepted == max) ifTrue(max_value := x value));
		peers select(k, 
			peers at(k) proposalNumber == responded) keys foreach(peer, 
		  		p := peers at(peer); 
				r := "http://#{peer}/accept/" interpolate;
				s := "#{r}#{p proposalNumber}/#{max_value}" interpolate;
		  		out(s)))
	nack := method(p, "nack #{p}" interpolate)
	start := method(loop(wait(Random value(8,16) ceil); 
		n := Date now asNumber ceil + Random value(4,32) ceil;
		peers values foreach(p, p proposalNumber = n);
		coroDo(prepare))))

Accepter := Object clone do(
	// need to save these to disk for restarts
	highAccepted := 0; highResponded := 0; currentValue := "none"
	prepare := method(n, peer,
		if(n > highResponded, 
			highResponded = n; coroDo(ack(peer)), 
			coroDo(nack(peer))))
	accept := method(proposal, value, 
		(proposal >= highResponded) ifTrue(
			highAccepted = proposal; currentValue = value; chose))
	ack := method(peer, r := "http://#{peer}:#{port}/ack/" interpolate;
		s := "#{r}#{highResponded}/" interpolate;
		t := "#{s}#{highAccepted}/#{currentValue}" interpolate;
		out(t))
	nack := method(peer, 
		s := "http://#{peer}:#{port}/nack/#{highResponded}" interpolate;
		out(s))
	chose := method(peers keys foreach(peer,
		s := "http://#{peer}/chose/#{currentValue}" interpolate;
		out(s))))

Learner := Object clone do(
	value := 0
	chose := method(v, (value == v) ifFalse(value = v; 
		"value: #{value}" interpolate println)))

WebRequest := Object clone do(
	handleSocket := method(aSocket,
		aSocket streamReadNextChunk
		request := aSocket readBuffer
		ip := aSocket ipAddress asString split(":") at(0)
		resource := request betweenSeq("GET ", " HTTP")
		debug ifTrue("< #{ip} #{resource}" interpolate println)
		resource = resource split("/")
		resource at(1) switch(
			"ack", Proposer ack(resource at(2) asNumber, 
				resource at(3) asNumber, resource at(4), ip),
			"nack", Proposer nack(resource at(2) asNumber),
			"prepare", Accepter prepare(resource at(2) asNumber, ip),
			"accept", Accepter accept(resource at(2) asNumber, 
				resource at(3)),
			"chose", Learner chose(resource at(2)), 
			resource println)
		aSocket close))
 
WebServer := Server clone do(
	setPort(4200)
	handleSocket := method(aSocket,
		coroDo(WebRequest clone handleSocket(aSocket))))

coroDo(WebServer start)
coroDo(loop(wait(60); "." println))
Proposer start
